package cn.money.youtube

import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

import java.util.concurrent.TimeUnit
import scala.util.Failure

case class YoutubeVideo(video_id: String
                        , trending_date: String
                        , title: String
                        , channel_title: String
                        , category_id: String
                        , publish_time: String
                        , tags: String
                        , views: Long
                        , likes: Long
                        , dislikes: Long
                        , comment_count: String
                        , thumbnail_link: String
                        , comments_disabled: Boolean
                        , ratings_disabled: Boolean
                        , video_error_or_removed: String
                        , description: String)

case class CategoryResult(categoryId:String
                          , totalLikes:Long
                          , totalDislikes:Long)

object DataSetAnalysis {
    def main(args: Array[String]): Unit = {
        val spark = SparkSession
            .builder()
            .appName("Youtube Analysis")
            .master("local[*]")
            .config("spark.sql.shuffle.partitions",1)
            .getOrCreate()

        import spark.sqlContext.implicits._

        // 读取CSV
        val youtubeVideoDF:DataFrame = spark.read.option("header", true).csv("""E:\05.git_project\dataset\USvideos.csv""")
        // 转换为DataSet
        youtubeVideoDF.printSchema()

        // 转换为Dataset[YoutubeVideo]
        val youtubeVideoDS = youtubeVideoDF.filter(row => {
            if(row.getString(7) != null && !row.getString(7).isBlank
                && row.getString(8) != null && !row.getString(8).isBlank
                && row.getString(9) != null && !row.getString(9).isBlank) {
                if(util.Try(row.getString(7).toLong).isSuccess
                    && util.Try(row.getString(8).toLong).isSuccess
                    && util.Try(row.getString(9).toLong).isSuccess) {
                    true
                }
                else {
                    false
                }
            }
            else {
                false
            }
        })
        .map(row => YoutubeVideo(row.getString(0)
            , row.getString(1)
            , row.getString(2)
            , row.getString(3)
            , row.getString(4)
            , row.getString(5)
            , row.getString(6)
            , row.getString(7).toLong
            , row.getString(8).toLong
            , row.getString(9).toLong
            , row.getString(10)
            , row.getString(11)
            , row.getString(12).toLowerCase().toBoolean
            , row.getString(13).toLowerCase().toBoolean
            , row.getString(14)
            , row.getString(15)
        ))

        youtubeVideoDS.groupByKey(_.category_id)
            .mapValues(y => CategoryResult(y.category_id, y.likes, y.dislikes))
            .reduceGroups{(cr1, cr2) => {
                CategoryResult(cr1.categoryId, cr1.totalLikes + cr2.totalLikes, cr1.totalDislikes + cr2.totalDislikes)
            }}
            // 只获取Value部分，key部分过滤掉
            .map(t => t._2)
            .toDF()
            .show()


        TimeUnit.HOURS.sleep(1)
    }
}
